package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TrafficPageViewBean;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple4;

import java.time.Duration;

/**
 * DWS 层是为 ADS 层服务的，通过对指标体系的分析，本节汇总表中需要有会话数、
 * 页面浏览数、浏览总时长、独立访客数、跳出会话数五个度量字段。本节的任务是统计这五
 * 个指标，并将维度和度量数据写入 ClickHouse 汇总表。
 */

/**
 *WebApp -> Nginx -> logFIle -> flume -> kafka -> flinKApp -> ClickHouse
 */
public class DWSTrafficVcChArIsNewPageViewWindowApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);  //kafka topic partitionNum = 4
        //checkpoint
        /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次
        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
         */

        //todo 统一数据格式
        //get topic
        String pageTopic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujdTopic = "dwd_traffic_user_jump_detail";
        String groupId = "dws_VcChArIsNew_pageView_window_zhu_2023";

        DataStreamSource<String> uvDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> ujDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(ujdTopic, groupId));
        DataStreamSource<String> pageDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId));

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUVDStream = uvDStream.map(data -> {
                    JSONObject jsonObject = JSON.parseObject(data);
                    JSONObject common = jsonObject.getJSONObject("common");
                    return new TrafficPageViewBean("", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            1L, 0L, 0L, 0L, 0L, jsonObject.getLong("ts"));
                }
        );

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUJDStream = ujDStream.map(data -> {
                    JSONObject jsonObject = JSON.parseObject(data);
                    JSONObject common = jsonObject.getJSONObject("common");
                    return new TrafficPageViewBean("", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            0L, 0L, 0L, 0L, 1L, jsonObject.getLong("ts"));
                }
        );

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithPageDStream = pageDStream.map(data -> {
                    JSONObject jsonObject = JSON.parseObject(data);
                    JSONObject common = jsonObject.getJSONObject("common");
            JSONObject page = jsonObject.getJSONObject("page");
            String lastPageId = page.getString("last_page_id");
            long sv = 0;
            if(lastPageId == null){
                sv = 1;
            }
            return new TrafficPageViewBean("", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            0L, sv, 1L, page.getLong("during_time"), 0L, jsonObject.getLong("ts"));
                }
        );
        //todo union
        DataStream<TrafficPageViewBean> unionDStream = trafficPageViewWithUVDStream.union(trafficPageViewWithUJDStream, trafficPageViewWithPageDStream);
        //todo 提取时间时间
        SingleOutputStreamOperator<TrafficPageViewBean> trafficViewWithWaterMarkDStream =
                unionDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(14))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                        return trafficPageViewBean.getTs();
                    }
                }));

        //todo 分组开创聚合
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowedStream = trafficViewWithWaterMarkDStream.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {
                return new Tuple4<>(value.getVc(), value.getCh(), value.getAr(), value.getIsNew());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)));//WindowAssigner 滚动窗口

        // aggregate reduce process   结合windowFunction
        SingleOutputStreamOperator<Object> resultDStream = windowedStream.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                     @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                },
                new WindowFunction<TrafficPageViewBean, Object, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4,
                                      TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable,
                                      Collector<Object> collector) throws Exception {
                        TrafficPageViewBean next = iterable.iterator().next();
                        next.setStt(DataFormatUtil.toYmdHms(timeWindow.getStart()));
                        next.setEdt(DataFormatUtil.toYmdHms(timeWindow.getEnd()));
                        //获取数据源头补充window 信息 并输出
                        collector.collect(next);
                    }
                });

        //todo 写入ClickHouse
        resultDStream.print(">>>>>");
        resultDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction("" +
                "insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(" +
                "?,?,?,?,?,?,?,?,?,?,?,?)"));

        //todo execute
        streamExecutionEnvironment.execute("DwsTrafficVcChArIsNewPageViewWindow");
    }
}
